package com.hanxiaozhang.mysql;

import com.hanxiaozhang.AbstractDistributedLock;
import com.hanxiaozhang.exception.DistributedLockException;
import lombok.extern.slf4j.Slf4j;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Timestamp;
import java.util.Arrays;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import static java.util.concurrent.Executors.*;

/**
 * 〈一句话功能简述〉<br>
 * 〈MySQL分布式锁〉
 * <p>
 * InnoDB引擎下，使用for update排他锁完成分布式锁。
 * 需要注意：
 * lock_key 字段要有索引，不然for update 会升级成表锁。
 * for update特点：
 * 1. 只允许获取锁的事务对数据进行操作（更新，删除）。
 * 2. 其他事务想要对相同的数据再次加锁的时，会进行到阻塞状态。
 * 建表语句：
 * CREATE TABLE `distribute_lock` (
 * `id` bigint  NOT NULL AUTO_INCREMENT,
 * `lock_key` varchar(100) NOT NULL,
 * `create_time` timestamp NULL DEFAULT NULL,
 * PRIMARY KEY (`id`),
 * UNIQUE KEY `lock_key` (`lock_key`)
 * ) ENGINE=InnoDB;
 *
 * @author hanxinghua
 * @create 2022/9/9
 * @since 1.0.0
 */
@Slf4j
public class MysqlDistributedLock extends AbstractDistributedLock {

    private static final String SELECT_SQL = "select * from distribute_lock where `lock_key`=? for update";
    private static final String INSERT_SQL = "insert into distribute_lock(`lock_key`，`create_time`) values(?,?)";

    private DataSource dataSource;

    private Connection connection;

    private ExecutorService executor = newSingleThreadExecutor();

    public MysqlDistributedLock(String lockKey, DataSource dataSource) {
        super(lockKey);
        this.dataSource = dataSource;
    }

    @Override
    public void lock() throws DistributedLockException {
        PreparedStatement statement = null;
        ResultSet resultSet = null;
        try {
            for (; ; ) {
                connection = dataSource.getConnection();
                connection.setAutoCommit(false);
                statement = connection.prepareStatement(SELECT_SQL);
                statement.setString(1, lockKey);
                resultSet = statement.executeQuery();
                // 存在加锁记录，并且加锁成功，则直接return
                if (resultSet.next()) {
                    return;
                }
                close(resultSet, statement, connection);
                // 锁记录不存在创建
                Connection insertConnection = dataSource.getConnection();
                PreparedStatement insertStatement = null;
                try {
                    insertStatement = insertConnection.prepareStatement(INSERT_SQL);
                    insertStatement.setString(1, lockKey);
                    insertStatement.setTimestamp(2, new Timestamp(System.currentTimeMillis()));
                    if (insertStatement.executeUpdate() == 1) {
                        log.info(lockKey + "的锁记录创建成功！");
                    }
                } catch (Exception e) {
                    throw new DistributedLockException("lockKey是 " + lockKey + " 加锁时，创建锁记录异常: ", e);
                } finally {
                    close(insertStatement, insertConnection);
                }
            }
        } catch (Exception e) {
            throw new DistributedLockException("lockKey是 " + lockKey + " 加锁异常: ", e);
        } finally {
            close(resultSet, statement);
        }
    }


    @Override
    public Boolean tryLock(long time, TimeUnit unit) throws DistributedLockException {
        final Future<Boolean> future = executor.submit(() -> {
            try {
                lock();
                return true;
            } catch (Exception e) {
                throw new DistributedLockException("lockKey是 " + lockKey + " 尝试加锁异常: ", e);
            }
        });
        try {
            if (null == future.get(time, unit)) {
                future.cancel(true);
                return false;
            }
            return true;
        } catch (Exception e) {
            return false;
        }
    }


    @Override
    public void unlock() throws DistributedLockException {
        try {
            connection.commit();
            close(connection);
        } catch (Exception e) {
            throw new DistributedLockException("lockKey是 " + lockKey + " 解锁异常: ", e);
        }
    }


    /**
     * 关闭链接
     *
     * @param closeables
     */
    private void close(AutoCloseable... closeables) {
        Arrays.stream(closeables).forEach(closeable -> {
            if (null != closeable) {
                try {
                    closeable.close();
                } catch (Exception e) {
                    log.error("close关闭异常：{}", e.getMessage());
                    throw new DistributedLockException(e);
                }
            }
        });
    }

}
